package com.xahj.bd2104.spark.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
 * Created with IntelliJ IDEA.
 * Author: Amos
 * E-mail: amos@amoscloud.com
 * Date: 2021/9/13
 * Time: 9:47
 * Description: 
 */
object SparkCoreDemo3_创建RDD {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("Demo3")
      .setMaster("local[2]")
    val sc = new SparkContext(conf)
    //RDD的操作

    // 1. 创建RDD
    //   SparkContext =>  RDD
    //  1.1 使用集合创建RDD
    //  1.1.1 makeRDD
    //  def makeRDD[T: ClassTag](
    //      seq: Seq[T],  传入集合作为元素来源
    //      numSlices: Int = defaultParallelism) // 当前生成RDD时，RDD的分区数
    //      : RDD[T] = 返回的RDD中的元素就是集合中的元素

    //  在创建RDD时可以指定分区数，Spark会为RDD的每个分区单独启动一个task进行并行运算
    //   分区数量与线程(task)数量一一对应
    //   Spark允许开发者随时调整RDD的分区数，以达到调整并发度的目的

    val rdd: RDD[Int] = sc.makeRDD(1 to 10)
    //  1.1.2 parallelize
    // makeRDD本质上就是封装了parallelize
    val rdd1: RDD[Int] = sc.parallelize(1 to 10)

    //  1.2 读取文件创建RDD

    // 1.2.1 文本文件
    //  textFile将文本文件按行读取到RDD中  每行作为一个String类型的元素
    //    val rdd2:RDD[String] = sc.textFile("")

    // 1.2.2 序列文件
    //                                sc.sequenceFile[K的反序列化类型,V反序列化类型]
    val rdd3: RDD[(String, String)] = sc.sequenceFile[String, String]("C:\\Users\\Amos\\Desktop\\output2\\part-r-00000")

    rdd3.flatMap(_._2.split("\n"))
      .map(x => (x.split(" ")(0), 1))
      .reduceByKey(_ + _)
      .sortBy(_._2, false)
      .repartition(1)
    //      .foreach(println)

    // 1.2.3 对象文件
    val rdd4 = sc.makeRDD(1 to 10)
    //    rdd4.saveAsObjectFile("C:\\Users\\Amos\\Desktop\\objectFileOut")

    val rdd5 = sc.objectFile("C:\\Users\\Amos\\Desktop\\objectFileOut")
    rdd5.foreach(println)
  }
}
